[API实践]基础聊天机器人应用实现

本节目标

搭建一个完整的聊天机器人应用
掌握基于DeepSeek API的对话系统设计
学习多轮对话的实现方法
了解用户体验优化的关键点
实现简单的错误处理和异常情况应对

项目概述

在本节中，我们将实现一个基础但功能完整的聊天机器人应用。这个应用将具备以下功能：

支持与用户进行自然语言对话
维护对话历史，实现上下文理解
处理不同类型的用户查询
提供流式输出，增强用户体验
实现简单的错误处理机制

这个项目将为我们后续学习更复杂的AI应用奠定基础。

技术准备

所需工具和库

Python 3.8+
OpenAI Python库（用于调用DeepSeek API）
终端交互或简单Web界面
环境变量管理（用于安全存储API密钥）

开发环境设置

首先，让我们创建一个虚拟环境并安装必要的依赖：

bash

# 创建并激活虚拟环境
python -m venv chatbot-env
source chatbot-env/bin/activate  # Linux/Mac
# 或 chatbot-env\Scripts\activate  # Windows

# 安装依赖
pip install openai python-dotenv

项目基础结构

创建一个基础的项目结构：

chatbot/
├── .env                 # 环境变量配置
├── chatbot.py           # 聊天机器人核心代码
├── config.py            # 配置管理
└── README.md            # 项目说明文档

环境配置文件

创建.env文件存储配置和密钥：

DEEPSEEK_API_KEY=your_api_key_here
DEEPSEEK_BASE_URL=https://api.deepseek.com

配置管理模块

创建config.py，用于管理配置信息：

python

import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# API设置
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com")

# 模型设置
MODEL_NAME = "deepseek-chat"

# 默认生成参数
DEFAULT_PARAMS = {
    "temperature": 0.7,
    "max_tokens": 800,
    "top_p": 0.95,
}

# 系统提示词
SYSTEM_PROMPT = """你是一位友好、专业的AI助手，能够回答用户的各种问题。
请提供准确、有帮助且符合道德的回答。
如果遇到不确定的问题，请坦诚表明你的局限性，而不是提供可能不准确的信息。
"""

聊天机器人核心实现

现在，让我们开始实现chatbot.py，这是我们聊天机器人的核心文件：

python

import sys
import time
from openai import OpenAI
import config

class ChatBot:
    def __init__(self):
        """初始化聊天机器人"""
        # 初始化OpenAI客户端
        self.client = OpenAI(
            api_key=config.DEEPSEEK_API_KEY,
            base_url=config.DEEPSEEK_BASE_URL
        )
        
        # 初始化对话历史
        self.messages = [
            {"role": "system", "content": config.SYSTEM_PROMPT}
        ]
        
        # 错误重试次数
        self.max_retries = 3
    
    def generate_response(self, user_input, stream=True):
        """生成对用户输入的回复"""
        # 添加用户输入到对话历史
        self.messages.append({"role": "user", "content": user_input})
        
        try:
            # 准备请求参数
            params = {
                "model": config.MODEL_NAME,
                "messages": self.messages,
                "stream": stream,
                **config.DEFAULT_PARAMS
            }
            
            # 调用API
            if stream:
                return self._stream_response(params)
            else:
                return self._standard_response(params)
                
        except Exception as e:
            error_msg = f"发生错误: {str(e)}"
            print(error_msg)
            return "很抱歉，我遇到了一些问题，无法生成回复。请稍后再试。"
    
    def _standard_response(self, params):
        """处理标准（非流式）响应"""
        retries = 0
        while retries < self.max_retries:
            try:
                response = self.client.chat.completions.create(**params)
                reply = response.choices[0].message.content
                
                # 将助手回复添加到对话历史
                self.messages.append({"role": "assistant", "content": reply})
                return reply
                
            except Exception as e:
                retries += 1
                if retries == self.max_retries:
                    raise
                # 简单的指数退避重试策略
                time.sleep(2 ** retries)
    
    def _stream_response(self, params):
        """处理流式响应"""
        retries = 0
        while retries < self.max_retries:
            try:
                full_response = ""
                response_stream = self.client.chat.completions.create(**params)
                
                # 一块一块地处理流式响应
                for chunk in response_stream:
                    if chunk.choices[0].delta.content:
                        content = chunk.choices[0].delta.content
                        full_response += content
                        print(content, end="", flush=True)
                
                print()  # 添加换行
                
                # 将完整回复添加到对话历史
                self.messages.append({"role": "assistant", "content": full_response})
                return full_response
                
            except Exception as e:
                retries += 1
                if retries == self.max_retries:
                    raise
                time.sleep(2 ** retries)
    
    def trim_conversation_history(self, max_turns=10):
        """裁剪对话历史，避免上下文过长"""
        # 保留system消息
        system_messages = [m for m in self.messages if m["role"] == "system"]
        # 获取非system消息
        other_messages = [m for m in self.messages if m["role"] != "system"]
        
        # 如果对话轮次超过最大值，只保留最近的几轮
        if len(other_messages) > max_turns * 2:  # 每轮包含用户和助手各一条消息
            other_messages = other_messages[-(max_turns * 2):]
        
        # 重建消息历史
        self.messages = system_messages + other_messages
    
    def clear_history(self):
        """清除对话历史，只保留system消息"""
        self.messages = [m for m in self.messages if m["role"] == "system"]
        if not self.messages:
            # 确保至少有一个system消息
            self.messages.append({"role": "system", "content": config.SYSTEM_PROMPT})

def main():
    print("欢迎使用DeepSeek聊天机器人！")
    print("输入'退出'或'exit'结束对话。")
    print("输入'清空'或'clear'清除对话历史。")
    print("-" * 50)
    
    bot = ChatBot()
    
    while True:
        user_input = input("\n你: ")
        
        # 检查特殊命令
        if user_input.lower() in ['退出', 'exit', 'quit']:
            print("感谢使用，再见！")
            break
        elif user_input.lower() in ['清空', 'clear']:
            bot.clear_history()
            print("对话历史已清空。")
            continue
        elif not user_input.strip():
            continue
            
        print("\nAI: ", end="")
        bot.generate_response(user_input)
        
        # 每5轮对话后裁剪历史，避免上下文过长
        bot.trim_conversation_history(max_turns=10)

if __name__ == "__main__":
    main()

运行和测试

保存上述文件后，通过以下命令运行聊天机器人：

bash

python chatbot.py

现在你应该能看到一个简单的命令行聊天界面，可以开始与AI进行对话了。

增强功能实现

让我们在基础版本上添加一些增强功能，使聊天机器人更加实用。

1. 对话模式切换

实现不同的对话模式，如"创意模式"、"专业模式"等，每种模式使用不同的参数设置：

python

# 在config.py中添加不同模式的参数
CHAT_MODES = {
    "standard": {
        "temperature": 0.7,
        "max_tokens": 800,
        "system_prompt": "你是一位友好、专业的AI助手，能够回答用户的各种问题。"
    },
    "creative": {
        "temperature": 1.2,
        "max_tokens": 1000,
        "system_prompt": "你是一位极具创造力的AI，擅长生成有趣、创新的内容。"
    },
    "precise": {
        "temperature": 0.2,
        "max_tokens": 800,
        "system_prompt": "你是一位追求精确和简洁的AI，提供准确、直接的回答，避免不必要的细节。"
    }
}

然后在ChatBot类中添加切换模式的方法：

python

def switch_mode(self, mode_name):
    """切换聊天模式"""
    if mode_name not in config.CHAT_MODES:
        return f"未知模式: {mode_name}。可用模式: {', '.join(config.CHAT_MODES.keys())}"
    
    mode = config.CHAT_MODES[mode_name]
    
    # 更新参数
    self.current_mode = mode_name
    self.current_params = {
        "temperature": mode["temperature"],
        "max_tokens": mode["max_tokens"],
    }
    
    # 更新system提示词
    self.clear_history()
    self.messages.append({"role": "system", "content": mode["system_prompt"]})
    
    return f"已切换到{mode_name}模式。"

2. 保存和加载对话

添加对话保存和加载功能：

python

import json
from datetime import datetime

def save_conversation(self, filename=None):
    """保存当前对话到文件"""
    if filename is None:
        # 使用时间戳创建默认文件名
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"conversation_{timestamp}.json"
    
    # 创建对话记录
    conversation_data = {
        "timestamp": datetime.now().isoformat(),
        "mode": getattr(self, "current_mode", "standard"),
        "messages": self.messages
    }
    
    # 保存到文件
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(conversation_data, f, ensure_ascii=False, indent=2)
    
    return filename

def load_conversation(self, filename):
    """从文件加载对话历史"""
    try:
        with open(filename, 'r', encoding='utf-8') as f:
            conversation_data = json.load(f)
        
        # 加载对话历史
        self.messages = conversation_data["messages"]
        
        # 恢复模式设置
        if "mode" in conversation_data:
            self.switch_mode(conversation_data["mode"])
        
        return True
    except Exception as e:
        print(f"加载对话失败: {str(e)}")
        return False

3. 特殊回答处理

添加针对特定问题的快速回应，无需调用API：

python

def get_quick_response(self, user_input):
    """检查是否可以提供快速回应，无需调用API"""
    # 转换为小写并移除额外空白
    query = user_input.lower().strip()
    
    # 快速回应字典
    quick_responses = {
        "你好": "你好！有什么我可以帮助你的吗？",
        "你的名字是什么": "我是基于DeepSeek API的AI助手。",
        "你是谁": "我是基于DeepSeek API的AI助手，随时准备回答您的问题。",
        "你能做什么": "我可以回答问题、生成内容、提供信息和建议等。有什么我可以帮助你的吗？",
        "今天几号": f"今天是{datetime.now().strftime('%Y年%m月%d日')}。",
        "现在几点": f"现在是{datetime.now().strftime('%H:%M:%S')}。",
    }
    
    # 检查是否有匹配的快速回应
    for key, response in quick_responses.items():
        if key in query:
            # 仍然添加到对话历史
            self.messages.append({"role": "user", "content": user_input})
            self.messages.append({"role": "assistant", "content": response})
            return response
    
    # 没有匹配的快速回应
    return None

4. 改进主函数

更新主函数，集成新功能：

python

def main():
    print("欢迎使用DeepSeek聊天机器人！")
    print("输入'退出'或'exit'结束对话。")
    print("输入'清空'或'clear'清除对话历史。")
    print("输入'模式'或'mode'后跟模式名切换模式（standard/creative/precise）。")
    print("输入'保存'或'save'保存当前对话。")
    print("输入'加载'或'load'后跟文件名加载对话。")
    print("-" * 50)
    
    bot = ChatBot()
    
    while True:
        user_input = input("\n你: ")
        
        # 检查特殊命令
        if user_input.lower() in ['退出', 'exit', 'quit']:
            print("感谢使用，再见！")
            break
        elif user_input.lower() in ['清空', 'clear']:
            bot.clear_history()
            print("对话历史已清空。")
            continue
        elif user_input.lower().startswith(('模式 ', 'mode ')):
            mode = user_input.split(' ', 1)[1].strip()
            result = bot.switch_mode(mode)
            print(result)
            continue
        elif user_input.lower() in ['保存', 'save']:
            filename = bot.save_conversation()
            print(f"对话已保存到: {filename}")
            continue
        elif user_input.lower().startswith(('加载 ', 'load ')):
            filename = user_input.split(' ', 1)[1].strip()
            if bot.load_conversation(filename):
                print(f"已成功加载对话: {filename}")
            else:
                print(f"加载对话失败。")
            continue
        elif not user_input.strip():
            continue
        
        # 尝试获取快速回应
        quick_response = bot.get_quick_response(user_input)
        if quick_response:
            print(f"\nAI: {quick_response}")
            continue
            
        # 调用API生成回复
        print("\nAI: ", end="")
        bot.generate_response(user_input)
        
        # 每10轮对话后裁剪历史，避免上下文过长
        bot.trim_conversation_history(max_turns=10)

优化用户体验

加载中动画

为了在等待API回复时提供更好的用户体验，我们可以添加一个简单的加载动画：

python

import threading
import itertools
import time
import sys

def loading_animation():
    """显示加载动画，直到停止标志被设置"""
    for c in itertools.cycle(['|', '/', '-', '\\']):
        if getattr(threading.current_thread(), "stop_loading", False):
            break
        sys.stdout.write('\r正在思考中 ' + c)
        sys.stdout.flush()
        time.sleep(0.1)
    sys.stdout.write('\r                   \r')

# 在generate_response方法中使用
def generate_response(self, user_input, stream=True):
    """生成对用户输入的回复"""
    self.messages.append({"role": "user", "content": user_input})
    
    if not stream:
        # 启动加载动画
        loading_thread = threading.Thread(target=loading_animation)
        loading_thread.start()
    
    try:
        # 准备请求参数
        params = {
            "model": config.MODEL_NAME,
            "messages": self.messages,
            "stream": stream,
            **config.DEFAULT_PARAMS
        }
        
        # 调用API
        if stream:
            return self._stream_response(params)
        else:
            result = self._standard_response(params)
            # 停止加载动画
            loading_thread.stop_loading = True
            loading_thread.join()
            return result
            
    except Exception as e:
        if not stream:
            # 停止加载动画
            loading_thread.stop_loading = True
            loading_thread.join()
        
        error_msg = f"发生错误: {str(e)}"
        print(error_msg)
        return "很抱歉，我遇到了一些问题，无法生成回复。请稍后再试。"

完整的聊天机器人应用

将以上所有功能组合在一起，我们就有了一个功能较为完整的聊天机器人应用。完整的代码可以在项目文件中查看。

项目扩展思路

这个基础聊天机器人可以通过以下方式进一步扩展：

添加Web界面：使用Flask或Streamlit创建Web界面
集成语音识别和合成：添加语音交互功能
添加知识库：结合本地文档或数据库，使机器人能回答特定领域问题
支持多语言：添加语言检测和翻译功能
个性化定制：允许用户保存偏好设置和对话风格

本节小结

我们实现了一个基于DeepSeek API的完整聊天机器人应用
掌握了对话历史管理和多轮对话实现方法
学习了错误处理和重试机制的实现
添加了模式切换、对话保存/加载等增强功能
探讨了优化用户体验的方法和项目扩展思路

实践任务

运行并测试基础聊天机器人应用
实现对话模式切换和保存/加载功能
为聊天机器人添加至少一个新功能（如命令解析、特定领域提示词等）
尝试优化用户体验，如添加颜色输出或简单的GUI界面

AI自动生成（Roo Code实战）

使用Roo Code的提示词：

在bots1的目录中创建一个基于deepseek api的聊天机器人。

可以通过.env的配置文件，设置DEEPSEEK_APIKEY与baseUrl。

在config.py中配置temperature与max_tokens，默认模型deepseek-chat，还有默认的系统prompt:

```
你是一位友好、专业的AI助手，能够回答用户的各种问题。
请提供准确、有帮助且符合道德的回答。
如果遇到不确定的问题，请坦诚表明你的局限性，而不是提供可能不准确的信息。
```

这个机器人可以进行多轮会话，当输入quit或者exit，退出程序；当输入清空或者clear，清空会话历史；当会话的轮数（user+assistant）超过10轮（20个）时，删除之前的user+assistant的会话历史，只保留system+10轮消息（user+assistant回复）

小游戏：

我们来玩一个猜数字的游戏, 你随机一个数,我来猜,这个数的范围是0-10, 我猜了之后,你要告诉我是大了还是小了,只能猜3次.

[API实践]基础聊天机器人应用实现 ​

本节目标 ​

项目概述 ​

技术准备 ​

所需工具和库 ​

开发环境设置 ​

项目基础结构 ​

环境配置文件 ​

配置管理模块 ​

聊天机器人核心实现 ​

运行和测试 ​

增强功能实现 ​

1. 对话模式切换 ​

2. 保存和加载对话 ​

3. 特殊回答处理 ​

4. 改进主函数 ​

优化用户体验 ​

加载中动画 ​

完整的聊天机器人应用 ​

项目扩展思路 ​

本节小结 ​

实践任务 ​

AI自动生成（Roo Code实战） ​

[API实践]基础聊天机器人应用实现

本节目标

项目概述

技术准备

所需工具和库

开发环境设置

项目基础结构

环境配置文件

配置管理模块

聊天机器人核心实现

运行和测试

增强功能实现

1. 对话模式切换

2. 保存和加载对话

3. 特殊回答处理

4. 改进主函数

优化用户体验

加载中动画

完整的聊天机器人应用

项目扩展思路

本节小结

实践任务

AI自动生成（Roo Code实战）